[SPARK-58322][CORE] Verify executor app ID at registration time#57525
[SPARK-58322][CORE] Verify executor app ID at registration time#57525wangyum wants to merge 2 commits into
Conversation
|
Thank you @wangyum! |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Two application-identity races can still allow an executor to register with the wrong driver. The inline comments describe the startup-cached missing-ID case and the config-fetch-versus-registration connection race.
| } | ||
|
|
||
| val cfg = driver.askSync[SparkAppConfig](RetrieveSparkAppConfig(arguments.resourceProfileId)) | ||
| verifyAppId(cfg.sparkProperties, arguments.appId) |
There was a problem hiding this comment.
[P1] Validate the application ID on the registration connection
This checks the driver identity only on driverPropsFetcher, but that RPC environment is shut down immediately afterwards. The executor subsequently creates a new SparkEnv, resolves driverUrl again in CoarseGrainedExecutorBackend.onStart, and sends RegisterExecutor on a different connection. RegisterExecutor contains no application ID, and DriverEndpoint accepts an otherwise valid, previously unused executor ID. Consequently, if the verified driver releases its port and a different application's driver binds the same address between config retrieval and registration, the executor still registers with the wrong application despite this check. This reproduces the port-reuse/data-corruption scenario the PR is intended to prevent under the default RPC authentication settings; the Kubernetes backend has the same two-connection sequence. Please include the executor's application ID in RegisterExecutor and validate it in DriverEndpoint, or otherwise bind the identity check to the connection used for registration. Add a regression test that swaps the driver between config retrieval and registration.
There was a problem hiding this comment.
Thank you. fix this issue
| private[spark] def verifyAppId( | ||
| sparkProperties: Seq[(String, String)], | ||
| executorAppId: String): Unit = { | ||
| sparkProperties.find(_._1 == "spark.app.id").map(_._2).foreach { driverAppId => |
There was a problem hiding this comment.
[P1] Do not accept an application ID missing from the driver's cached configuration
DriverEndpoint.sparkProperties is an immutable lazy val, initialized by the first RetrieveSparkAppConfig request. However, SparkContext calls _taskScheduler.start() before _conf.set("spark.app.id", _applicationId), and the standalone scheduler can launch an executor during that startup. If the first executor fetches its config in this window, the cached driver properties omit spark.app.id for the entire application lifetime. This .find(...).foreach then silently succeeds for every later executor, including one belonging to another application, so the proposed protection is permanently disabled. The new appId verification no-op when spark.app.id absent test actually codifies this failure. Please obtain the finalized application ID from an authoritative driver/scheduler source and reject an unverifiable identity, ideally in the registration handler; add a regression test that fetches configuration before the driver publishes its application ID.
There was a problem hiding this comment.
Thank you. fix this issue
…tion Add appId to RegisterExecutor message and validate it in DriverEndpoint against scheduler.sc.applicationId. This binds the identity check to the registration connection itself, preventing the config-fetch-vs-registration race where a wrong driver could bind the same port between config retrieval and registration. The check replaces the executor-side verifyAppId which ran on a transient RPC connection that was shut down before registration. The driver-side check is authoritative since scheduler.sc.applicationId is always set before any executor can register. Null appId is accepted for backward compatibility with custom/external cluster managers that construct RegisterExecutor without appId.
wangyum
changed the title
3 participants
What changes were proposed in this pull request?
This PR adds an
appIdfield to theRegisterExecutorRPC message and validates it inDriverEndpoint.receiveAndReplyagainstscheduler.sc.applicationId. The executor sendsenv.conf.getAppId(which is the--app-idlaunch argument) inRegisterExecutor. On mismatch, the driver rejects the registration with aSparkException.appIddefaults tonullinRegisterExecutorfor backward compatibility with custom/external cluster managers that construct the message without it.Option(appId).exists(_ != scheduler.sc.applicationId)cleanly skips the check whenappIdisnull.Why are the changes needed?
When a driver hits a fatal error (e.g., OOM) that kills the RPC dispatcher thread,
SparkContext.stop()releases the driver's RPC port while the driver process (and in YARN cluster mode, the ApplicationMaster) can remain in a zombie state — still heartbeating to the resource manager but effectively dead. During this window, the freed port can be rebound by another driver on the same host, and newly launched executors can connect to the wrong driver and register to the wrong application, risking data corruption.Does this PR introduce any user-facing change?
Yes. In the rare case of an app-ID mismatch at registration, the executor will now be rejected by the driver instead of registering to the wrong application. The error message is:
No config or API changes. No impact on normal operation. External cluster managers that do not set
appIdinRegisterExecutorare unaffected (the check is skipped).How was this patch tested?
Unit tests in
CoarseGrainedSchedulerBackendSuite:SPARK-58322: reject RegisterExecutor with mismatched app ID— verifies mismatch is rejectedSPARK-58322: accept RegisterExecutor with matching app ID— verifies match is acceptedSPARK-58322: accept RegisterExecutor with null app ID for backward compatibility— verifies nullappIdis acceptedSPARK-58322: reject RegisterExecutor after driver swap between config fetch and registration— simulates the port-reuse scenario: config fetched from a fake driver A, thenRegisterExecutorsent to real driver B, verifies rejectionWas this patch authored or co-authored using generative AI tooling?
Generated-by: GLM 5.2.